package com.daviteq.ems.dao.cassandra;

import java.io.IOException;

import org.apache.log4j.Logger;

import com.datastax.driver.core.Cluster;
import com.datastax.driver.core.HostDistance;
import com.datastax.driver.core.PoolingOptions;
import com.datastax.driver.core.Session;
import com.datastax.driver.core.SocketOptions;
import com.datastax.driver.mapping.MappingSession;
import com.daviteq.ems.utils.CassandraConfig;
import com.daviteq.ems.utils.EMSConfig;

public class SessionFatory {

    private final static Logger log = Logger.getLogger(SessionFatory.class);
    
    protected CassandraConfig cfg;
    
    private SessionFatory() {
        EMSConfig emsCfg = EMSConfig.getInstance(); 
        try {
            cfg = emsCfg.loadGlobalConfig();
        } catch (IOException e) {
            log.error(e);
        }
    }

    private static class Loader {

        private static final SessionFatory INSTANCE = new SessionFatory();
    }

    public static SessionFatory getInstance() {
        return Loader.INSTANCE;
    }

    private Cluster cluster;
    private Session session;
    private MappingSession mSession;

    public Session getSession() {
        if (session == null) {
            connect();
        }
        return session;
    }

    public MappingSession getMappingSession() {
        if (session == null) {
            connect();
        }
        return mSession;
    }

    public String getKeyspace() {
        return cfg.getKeyspace();
    }

    /** only 1 thread is permitted to open connection */
    protected synchronized void connect() {

        if (session == null) {

            PoolingOptions pools = new PoolingOptions();
            
            pools.setMaxSimultaneousRequestsPerConnectionThreshold(
                    HostDistance.LOCAL, cfg.getMaxSimultaneousRequests());
            
            pools.setCoreConnectionsPerHost(HostDistance.LOCAL,
                    cfg.getCoreConnectionsPerHost());
            
            pools.setMaxConnectionsPerHost(HostDistance.LOCAL,
                    cfg.getMaxConnectionPerHost());
            
//            pools.setMaxSimultaneousRequestsPerConnectionThreshold(
//                    HostDistance.REMOTE, cfg.getMaxSimultaneousRequests());
//            
//            pools.setCoreConnectionsPerHost(HostDistance.REMOTE,
//                    cfg.getCoreConnectionsPerHost());
//            
//            pools.setMaxConnectionsPerHost(HostDistance.REMOTE,
//                    cfg.getMaxConnectionPerHost());

            SocketOptions socket = new SocketOptions();
            
            socket.setConnectTimeoutMillis(cfg.getConnectTimeoutMillis());
            
            socket.setKeepAlive(cfg.isKeepAlive());
            
            socket.setReadTimeoutMillis(cfg.getReadTimeoutMillis());
            
            cluster = Cluster
                    .builder()
                    .withPoolingOptions(pools)
                    .withSocketOptions(socket)
                    .addContactPoint(cfg.getNode())
                    .withPort(cfg.getPort())
                    .build();
            
            
            
/** begin debug **/
/*new Thread(new Runnable() {

    public void run() {
        int n = 0;
        while(n < 20) {
            
            PoolingOptions pool = cluster.getConfiguration().getPoolingOptions();
            
            int coreConnPerLocalHost = pool.getCoreConnectionsPerHost(HostDistance.LOCAL);
            int coreConnPerRemoteHost = pool.getCoreConnectionsPerHost(HostDistance.REMOTE);
            int coreConnPerIgnoredHost = pool.getCoreConnectionsPerHost(HostDistance.IGNORED);
            
            int maxConnPerLocalHost = pool.getMaxConnectionsPerHost(HostDistance.LOCAL);
            int maxConnPerRemoteHost = pool.getMaxConnectionsPerHost(HostDistance.REMOTE);
            int maxConnPerIgnoredHost = pool.getMaxConnectionsPerHost(HostDistance.IGNORED);
            
            int maxSimulReqPerConnThresLocal = pool.getMaxSimultaneousRequestsPerConnectionThreshold(HostDistance.LOCAL);
            int maxSimulReqPerConnThresRemote = pool.getMaxSimultaneousRequestsPerConnectionThreshold(HostDistance.REMOTE);
            int maxSimulReqPerConnThresIgnored = pool.getMaxSimultaneousRequestsPerConnectionThreshold(HostDistance.IGNORED);
            
            int minSimulReqPerConnThresLocal = pool.getMinSimultaneousRequestsPerConnectionThreshold(HostDistance.LOCAL);
            int minSimulReqPerConnThresRemote = pool.getMinSimultaneousRequestsPerConnectionThreshold(HostDistance.REMOTE);
            int minSimulReqPerConnThresIgnored = pool.getMinSimultaneousRequestsPerConnectionThreshold(HostDistance.IGNORED);
            
            log.debug("coreConnPerLocalHost: " + coreConnPerLocalHost);
            log.debug("coreConnPerRemoteHost: " + coreConnPerRemoteHost);
            log.debug("coreConnPerIgnoredHost: " + coreConnPerIgnoredHost);
            
            log.debug("maxConnPerLocalHost: " + maxConnPerLocalHost);
            log.debug("maxConnPerRemoteHost: " + maxConnPerRemoteHost);
            log.debug("maxConnPerIgnoredHost: " + maxConnPerIgnoredHost);
            
            log.debug("maxSimulReqPerConnThresLocal: " + maxSimulReqPerConnThresLocal);
            log.debug("maxSimulReqPerConnThresRemote: " + maxSimulReqPerConnThresRemote);
            log.debug("maxSimulReqPerConnThresIgnored: " + maxSimulReqPerConnThresIgnored);
            
            log.debug("minSimulReqPerConnThresLocal: " + minSimulReqPerConnThresLocal);
            log.debug("minSimulReqPerConnThresRemote: " + minSimulReqPerConnThresRemote);
            log.debug("minSimulReqPerConnThresIgnored: " + minSimulReqPerConnThresIgnored);
            
            try {
                Thread.sleep(30000);
                n++;
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            
        }
        
    }
}).start();*/
/** end debug **/
            
            session = cluster.connect();
            session.execute("CREATE KEYSPACE IF NOT EXISTS "
                    + getKeyspace()
                    + " WITH REPLICATION = { 'class' : 'SimpleStrategy', 'replication_factor' : 1 }");

            mSession = new MappingSession(getKeyspace(), getSession());
        }
    }
    
}
